package mongo

import (
	"container/list"
	"core/common"
	"core/component/logger"
	"fmt"
	"github.com/globalsign/mgo"
	"github.com/globalsign/mgo/bson"
	"math/rand"
	"strings"
	"time"
)

func removeOneIp(index int, ipList []string) []string {
	if index < 0 || index >= len(ipList) {
		return ipList
	}
	for i := index; i < len(ipList)-1; i++ {
		ipList[i] = ipList[i+1]
	}
	return ipList[:len(ipList)-1]
}

func removeIpFromIpList(ip string, ipList []string) []string {
	for i := len(ipList) - 1; i >= 0; i-- {
		if ip == ipList[i] {
			ipList = removeOneIp(i, ipList)
		}
	}
	return ipList
}

type UpdateInfo struct {
	Collection       string
	Selector, Update interface{}
}

// MgoClient mongodb客户端连接
type MgoClient struct {
	dbName            string
	session           *mgo.Session
	callCounter       [31]uint64
	lastCountTime     time.Time
	usingURL          string        // 使用中的连接配置
	firstURL          string        // 建立连接的主配置
	bakURLs           []string      // 主配置建立连接异常后，用备用配置建立连接
	lastActiveTime    time.Time     // 连接最后一次活跃时间，用于判断连接是否有效
	KeepActiveInterv  time.Duration // 检测连接时间间隔
	KeepActiveTimeout time.Duration // 检测连接时间间隔
}

func rebuildURL(url string, windowSize, idx int) (string, []string) {
	var part1, part2, part3 string
	c := strings.Index(url, "@")
	if c == -1 {
		return url, nil
	}
	part1 = url[:c+1]
	subStr := url[c+1:]
	i := strings.Index(subStr, "/")
	if i == -1 {
		return url, nil
	}
	part2 = subStr[:i]
	part3 = subStr[i:]
	ips := strings.Split(part2, ",")
	ln := len(ips)
	if ln <= 1 {
		return url, nil
	}

	if windowSize <= 0 {
		windowSize = 2
	}

	if windowSize > ln {
		windowSize = ln
	}

	if idx < 0 {
		idx = 0
	}

	offset := idx % ln
	var newIps []string
	for i = 0; i < windowSize; i++ {
		if offset+i < ln {
			newIps = append(newIps, ips[offset+i])
		} else {
			newIps = append(newIps, ips[(offset+i)%ln])
		}
	}

	ipStr := strings.Join(newIps, ",")
	firstURL := part1 + ipStr + part3

	// remove the ips that has been used
	for i = 0; i < len(newIps); i++ {
		ip := newIps[i]
		ips = removeIpFromIpList(ip, ips)
	}
	restCnt := len(ips)
	if restCnt > 0 {
		// Shuffle the order
		rand.Seed(time.Now().UnixNano())
		rand.Shuffle(len(ips), func(i, j int) { ips[i], ips[j] = ips[j], ips[i] })

		// build bak urls list
		bakURLs := []string{}
		newIps = []string{}
		for i = 0; i < restCnt; i++ {
			ip := ips[i]
			newIps = append(newIps, ip)
			if (i+1)%windowSize == 0 || (i+1) == restCnt {
				ipStr = strings.Join(newIps, ",")
				URL := part1 + ipStr + part3
				bakURLs = append(bakURLs, URL)
				newIps = newIps[:0]
			}
		}
		return firstURL, bakURLs
	} else {
		return firstURL, nil
	}

}

// NewMgoClient 创建mongodb连接客户端
func NewMgoClient(dbName, url string, poolSize int, idx int) (*MgoClient, error) {
	var err error
	mgoClient := &MgoClient{}
	mgoURL, bakURLs := rebuildURL(url, poolSize, idx)
	mgoClient.session, err = mgo.Dial(mgoURL)
	if err != nil {
		return nil, err
	}
	mgoClient.dbName = dbName
	mgoClient.session.SetMode(mgo.Primary, true)
	mgoClient.session.SetPoolLimit(poolSize)

	// 处理动态备用连接
	mgoClient.firstURL = mgoURL
	mgoClient.bakURLs = bakURLs
	mgoClient.lastActiveTime = time.Now()
	mgoClient.KeepActiveInterv = 30 * time.Second
	mgoClient.KeepActiveTimeout = 4 * mgoClient.KeepActiveInterv
	mgoClient.usingURL = mgoClient.firstURL

	logger.Infof("[mongo] 创建mongodb客户端连接 URL:%v, 备用IPS:%v poolSize:%v", mgoURL, bakURLs, poolSize)
	return mgoClient, nil
}

func (mgoClient *MgoClient) DetectActiveTimer() {
	if time.Since(mgoClient.lastActiveTime) >= mgoClient.KeepActiveInterv {
		ms := mgoClient.session.Copy()
		e := ms.Ping()
		ms.Close()
		if e == nil {
			mgoClient.lastActiveTime = time.Now()
		} else if time.Since(mgoClient.lastActiveTime) >= mgoClient.KeepActiveTimeout {
			// lost two times ping pong, so try ...
			mgoClient._tryConnectWithBakURL()
		}
	}
}

func (mgoClient *MgoClient) _tryConnectWithBakURL() {
	if time.Since(mgoClient.lastActiveTime) >= mgoClient.KeepActiveTimeout {
		if mgoClient.bakURLs != nil && len(mgoClient.bakURLs) > 0 {
			// close old session ???
			nextURL := mgoClient.bakURLs[0]
			mgoClient.bakURLs = mgoClient.bakURLs[1:]

			prevSess := mgoClient.session
			var err error
			mgoClient.session, err = mgo.Dial(nextURL)
			if err != nil {
				logger.Errorf("_tryConnectWithBakURL with bakURL: %v error: %v",
					nextURL, err.Error())
				return
			}
			if prevSess != nil {
				prevSess.Close()
			}
			mgoClient.usingURL = nextURL
			mgoClient.session.SetMode(mgo.Primary, true)
			mgoClient.lastActiveTime = time.Now()
			logger.Infof("_tryConnectWithBakURL ok usingURL: %v, rest bakURLs:%v",
				mgoClient.usingURL, mgoClient.bakURLs)
		} else if mgoClient.usingURL != mgoClient.firstURL {
			// all bak urls failed, try rewind to first main URL
			prevSess := mgoClient.session
			var err error
			mgoClient.session, err = mgo.Dial(mgoClient.firstURL)
			if err != nil {
				logger.Errorf("_tryConnectWithBakURL rewind with firstURL: %v error: %v",
					mgoClient.firstURL, err.Error())
				return
			}
			if prevSess != nil {
				prevSess.Close()
			}
			mgoClient.usingURL = mgoClient.firstURL
			mgoClient.session.SetMode(mgo.Primary, true)
			mgoClient.lastActiveTime = time.Now()
			logger.Infof("_tryConnectWithBakURL rewind to firstURL ok: %v.",
				mgoClient.firstURL)
		}
	}
}

// Close 关闭连接seeesion
func (mgoClient *MgoClient) Close() {
	mgoClient.session.Close()
}

func (mgoClient *MgoClient) callCounterTick() {
	// 一分钟log一次访问统计结果
	if time.Since(mgoClient.lastCountTime) > 1*time.Minute {

		for idx, count := range mgoClient.callCounter {
			if count == 0 {
				continue
			}
			logger.Infof("[mongo] call counter idx:%v count:%v", common.CollIdxMap[uint16(idx)], count)
		}

		mgoClient.lastCountTime = time.Now()
	}
}

func (mgoClient *MgoClient) getSession(collection string) (*mgo.Session, *mgo.Collection) {
	ms := mgoClient.session.Copy()
	c := ms.DB(mgoClient.dbName).C(collection)
	ms.SetMode(mgo.Primary, true)

	// collIdx, ok := common.CollNameMap[collection]
	// if ok {
	// 	atomic.AddUint64(&mgoClient.callCounter[collIdx], 1)
	// }

	// mgoClient.callCounterTick()

	return ms, c
}

// IsEmpty 集合是否为空集合
func (mgoClient *MgoClient) IsEmpty(collection string) bool {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	count, err := c.Count()
	if err != nil {
		logger.Errorf("[DBOP] 数据库IsEmpty操作错误:%v", err)
		return false
	}
	return count == 0
}

// Count 返回结果集数量(查询结果) 对集合全部数据记录的统计请使用ColCount
func (mgoClient *MgoClient) Count(dcollection string, query interface{}) (int, error) {
	ms, c := mgoClient.getSession(dcollection)
	defer ms.Close()
	count, err := c.Find(query).Count()
	if err != nil {
		logger.Errorf("[DBOP] 数据库Count操作错误:%v", err)
		return 0, err
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return count, err
}

// ColCount 集合的数量统计
func (mgoClient *MgoClient) ColCount(collection string) (int, error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	count, err := c.Count()
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-Count操作错误:%v", mgoClient.dbName, collection, err)
		return 0, err
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return count, err
}

// Apply ...
func (mgoClient *MgoClient) Apply(collection string,
	query interface{}, change mgo.Change, result interface{}) (info *mgo.ChangeInfo, err error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	info, err = c.Find(query).Apply(change, result)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-Apply操作 query:%v 错误:%v", mgoClient.dbName, collection, query, err)
		return nil, err
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return info, err
}

// Insert 集合插入数据
func (mgoClient *MgoClient) Insert(collection string, docs ...interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	err := c.Insert(docs...)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-Insert操作错误:%v", mgoClient.dbName, collection, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}

	return err
}

// FindOne 查询一条记录
func (mgoClient *MgoClient) FindOne(collection string, query, selector, result interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	err := c.Find(query).Select(selector).One(result)
	if err != nil && err != mgo.ErrNotFound {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-FindOne操作 query:%v 错误:%v", mgoClient.dbName, collection, query, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// FindAll ...
func (mgoClient *MgoClient) FindAll(collection string, query, selector, result interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	err := c.Find(query).Select(selector).All(result)
	if err != nil && err != mgo.ErrNotFound {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-FindAll操作 query:%v 错误:%v", mgoClient.dbName, collection, query, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// FindPage 分页查询
func (mgoClient *MgoClient) FindPage(collection string, page, limit int, query, selector, result interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	err := c.Find(query).Select(selector).Skip(page * limit).Limit(limit).All(result)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-FindPage操作 query:%v 错误:%v", mgoClient.dbName, collection, query, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// FindIter ...
func (mgoClient *MgoClient) FindIter(db, collection string, query interface{}) *mgo.Iter {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	return c.Find(query).Iter()
}

// Update 更新
func (mgoClient *MgoClient) Update(collection string, selector, update interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	err := c.Update(selector, update)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-Update操作 condition:%v 错误:%v update:%v", mgoClient.dbName, collection, selector, err, update)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// Update2 更新  防止mgo HA期间写失败
func (mgoClient *MgoClient) Update2(collection string, selector, update interface{}, list *list.List) error {
	list.PushBack(UpdateInfo{collection, selector, update})
	for i := list.Front(); i != nil; {
		info := i.Value.(UpdateInfo)
		err := mgoClient.Update(info.Collection, info.Selector, info.Update)
		if err != nil {
			return err
		}
		tmp := i
		i = i.Next()
		list.Remove(tmp)
	}
	return nil
}

// UpdateFields 更新
func (mgoClient *MgoClient) UpdateFields(collection string, selector interface{}, fields ...interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	fieldsLen := len(fields)
	if fieldsLen%2 != 0 {
		err := fmt.Errorf("[DBOP] 数据库:%v集合:%v-UpdateFields操作 selector:%v 字段数量不匹配", mgoClient.dbName, collection, selector)
		logger.Error(err)
		return err
	}
	fieldsName := fields[:fieldsLen/2]
	fieldsData := fields[fieldsLen/2:]

	mdata := bson.M{}
	for idx, fname := range fieldsName {
		fieldName, ok := fname.(string)
		if !ok {
			continue
		}
		mdata[fieldName] = fieldsData[idx]
	}
	update := bson.M{"$set": mdata}
	err := c.Update(selector, update)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-UpdateFields操作 condition:%v fields:%v 错误:%v",
			mgoClient.dbName, collection, selector, fieldsName, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// Upsert 更新插入
func (mgoClient *MgoClient) Upsert(collection string, selector, update interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	_, err := c.Upsert(selector, update)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-Upsert操作 condition:%v 错误:%v", mgoClient.dbName, collection, selector, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// UpdateAll ...
func (mgoClient *MgoClient) UpdateAll(collection string, selector, update interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	_, err := c.UpdateAll(selector, update)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-UpdateAll操作 condition:%v 错误:%v", mgoClient.dbName, collection, selector, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// Remove ...
func (mgoClient *MgoClient) Remove(collection string, selector interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	err := c.Remove(selector)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-Remove操作 selector:%v 错误:%v", mgoClient.dbName, collection, selector, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}

	return err
}

// RemoveAll ...
func (mgoClient *MgoClient) RemoveAll(collection string, selector interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	_, err := c.RemoveAll(selector)
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-RemoveAll操作 selector:%v 错误:%v", mgoClient.dbName, collection, selector, err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// BulkInsert insert one or multi documents
func (mgoClient *MgoClient) BulkInsert(collection string, docs ...interface{}) (*mgo.BulkResult, error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	bulk := c.Bulk()
	bulk.Insert(docs...)
	ret, err := bulk.Run()
	if err != nil {
		logger.Errorf("[DBOP] 数据库BulkInsert操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return ret, err
}

// BulkRemove ...
func (mgoClient *MgoClient) BulkRemove(collection string, selector ...interface{}) (*mgo.BulkResult, error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()

	bulk := c.Bulk()
	bulk.Remove(selector...)
	ret, err := bulk.Run()
	if err != nil {
		logger.Errorf("[DBOP] 数据库BulkRemove操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return ret, err
}

// BulkRemoveAll ...
func (mgoClient *MgoClient) BulkRemoveAll(collection string, selector ...interface{}) (*mgo.BulkResult, error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	bulk := c.Bulk()
	bulk.RemoveAll(selector...)
	ret, err := bulk.Run()
	if err != nil {
		logger.Errorf("[DBOP] 数据库BulkRemoveAll操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return ret, err
}

// BulkUpdate ...
func (mgoClient *MgoClient) BulkUpdate(collection string, pairs ...interface{}) (*mgo.BulkResult, error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	bulk := c.Bulk()
	bulk.Update(pairs...)
	ret, err := bulk.Run()
	if err != nil {
		logger.Errorf("[DBOP] 数据库:%v集合:%v-BulkUpdate操作 错误:%v update:%v", mgoClient.dbName, collection, err, pairs)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return ret, err
}

// BulkUpdateAll ...
func (mgoClient *MgoClient) BulkUpdateAll(collection string, pairs ...interface{}) (*mgo.BulkResult, error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	bulk := c.Bulk()
	bulk.UpdateAll(pairs...)
	ret, err := bulk.Run()
	if err != nil {
		logger.Errorf("[DBOP] 数据库BulkUpdateAll操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return ret, err
}

// BulkUpsert ...
func (mgoClient *MgoClient) BulkUpsert(collection string, pairs ...interface{}) (*mgo.BulkResult, error) {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	bulk := c.Bulk()
	bulk.Upsert(pairs...)
	ret, err := bulk.Run()
	if err != nil {
		logger.Errorf("[DBOP] 数据库BulkUpsert操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return ret, err
}

// PipeAll ...
func (mgoClient *MgoClient) PipeAll(collection string, pipeline, result interface{}, allowDiskUse bool) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	var pipe *mgo.Pipe
	if allowDiskUse {
		pipe = c.Pipe(pipeline).AllowDiskUse()
	} else {
		pipe = c.Pipe(pipeline)
	}
	err := pipe.All(result)
	if err != nil {
		logger.Errorf("[DBOP] 数据库PipAll操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// PipeOne ...
func (mgoClient *MgoClient) PipeOne(collection string, pipeline, result interface{}, allowDiskUse bool) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	var pipe *mgo.Pipe
	if allowDiskUse {
		pipe = c.Pipe(pipeline).AllowDiskUse()
	} else {
		pipe = c.Pipe(pipeline)
	}
	err := pipe.One(result)
	if err != nil {
		logger.Errorf("[DBOP] 数据库PipOne操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}

// PipeIter ...
func (mgoClient *MgoClient) PipeIter(collection string, pipeline interface{}, allowDiskUse bool) *mgo.Iter {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	var pipe *mgo.Pipe
	if allowDiskUse {
		pipe = c.Pipe(pipeline).AllowDiskUse()
	} else {
		pipe = c.Pipe(pipeline)
	}

	return pipe.Iter()

}

// UnWind ...
func (mgoClient *MgoClient) UnWind(collection, unwind string, match, selector interface{}) []bson.M {
	se, c := mgoClient.getSession(collection)
	defer se.Close()
	pipe := c.Pipe([]bson.M{{"$unwind": unwind},
		{"$match": match},
		{"$project": selector}})
	var resp []bson.M
	if err := pipe.All(&resp); err != nil {
		logger.Errorf("UnWind error %v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}

	return resp
}

// Explain ...
func (mgoClient *MgoClient) Explain(collection string, pipeline, result interface{}) error {
	ms, c := mgoClient.getSession(collection)
	defer ms.Close()
	pipe := c.Pipe(pipeline)
	err := pipe.Explain(result)
	if err != nil {
		logger.Errorf("[DBOP] 数据库Explain操作错误:%v", err)
	} else {
		mgoClient.lastActiveTime = time.Now()
	}
	return err
}
